package com.chb.flink

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

/**
 * 基于流计算的WordCount案例
 */
object StreamWordCount {
    def main(args: Array[String]): Unit = {
        //初始化Flink的Streaming（流计算）上下文执行环境
        val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        //导入隐式转换，建议写在这里，可以防止IDEA代码提示出错的问题
        import org.apache.flink.streaming.api.scala._

        //读取数据
        val stream: DataStream[String] = streamEnv.socketTextStream("10.0.0.201", 8888)
        //转换计算
        val result: DataStream[(String, Int)] = stream.flatMap(_.split(","))
            .map((_, 1))
            .keyBy(0)
            .sum(1)

        //打印结果到控制台
        result.print()
        //启动流式处理，如果没有该行代码上面的程序不会运行
        streamEnv.execute("wordcount")
    }
}